package com.bd03.streaminglearn.day0327

import org.apache.log4j.{Level, Logger}
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.ReceiverInputDStream
import org.apache.spark.streaming.{Seconds, StreamingContext}
import redis.clients.jedis.Jedis

object WordCountRedis {
  def main(args: Array[String]): Unit = {
    Logger.getLogger("org").setLevel(Level.WARN)
    val conf = new SparkConf().setAppName(this.getClass.getSimpleName).setMaster("local[*]")
    val ssc = new StreamingContext(conf,Seconds(2))
    val data: ReceiverInputDStream[String] = ssc.socketTextStream("hdp01",9999)
    val res = data.flatMap(t=>t.split(" ")).map((_,1)).reduceByKey(_+_)
    res.print()
    res.foreachRDD(rdd=>{
      rdd.foreachPartition(pat=>{
        val jedis = JedisPoolDemo.getResource()
        pat.foreach(t=>{
          jedis.hincrBy("wordcount",t._1,t._2)//有该key就修改,没有该key就直接set
        })
        jedis.close()
      })
    })

    ssc.start()
    ssc.awaitTermination()

  }
}
